#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
RedFire 跨数据库数据一致性检查工具
================================

功能特性:
- MySQL ↔ PostgreSQL 数据一致性验证
- 实时数据质量监控
- 自动修复建议
- 详细的一致性报告
- 支持大规模数据检查

Author: RedFire Team
Date: 2025-09-16
Version: v1.0.0
"""

import asyncio
import logging
from datetime import datetime, timedelta, date
from typing import Dict, List, Optional, Tuple, Any, Union
from dataclasses import dataclass, asdict
from enum import Enum
import json
import hashlib
import statistics
from collections import defaultdict, Counter
import aiomysql
import asyncpg
import numpy as np
import pandas as pd
from decimal import Decimal, getcontext
import traceback

# 设置高精度计算
getcontext().prec = 28

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)


class InconsistencyType(Enum):
    """不一致类型"""
    MISSING_RECORD = "missing_record"
    EXTRA_RECORD = "extra_record"
    DATA_MISMATCH = "data_mismatch"
    TYPE_MISMATCH = "type_mismatch"
    PRECISION_LOSS = "precision_loss"
    ENCODING_ERROR = "encoding_error"
    TIMESTAMP_DRIFT = "timestamp_drift"
    NULL_VALUE_MISMATCH = "null_value_mismatch"


class SeverityLevel(Enum):
    """严重性级别"""
    CRITICAL = "critical"  # 影响回测结果准确性
    HIGH = "high"         # 可能影响分析质量
    MEDIUM = "medium"     # 数据质量问题
    LOW = "low"          # 轻微不一致
    INFO = "info"        # 信息性提示


@dataclass
class InconsistencyRecord:
    """不一致记录"""
    inconsistency_type: InconsistencyType
    severity: SeverityLevel
    table_name: str
    primary_key: str
    field_name: str = None
    mysql_value: Any = None
    postgresql_value: Any = None
    description: str = ""
    detected_at: datetime = None
    
    def __post_init__(self):
        if self.detected_at is None:
            self.detected_at = datetime.now()


@dataclass
class ConsistencyReport:
    """一致性检查报告"""
    check_id: str
    table_name: str
    check_start_time: datetime
    check_end_time: datetime
    total_records_checked: int
    mysql_record_count: int
    postgresql_record_count: int
    inconsistencies: List[InconsistencyRecord]
    
    # 统计信息
    consistency_score: float = 0.0  # 0-100的一致性分数
    error_rate: float = 0.0
    
    def __post_init__(self):
        self._calculate_metrics()
    
    def _calculate_metrics(self):
        """计算指标"""
        if self.total_records_checked > 0:
            self.error_rate = len(self.inconsistencies) / self.total_records_checked
            self.consistency_score = max(0, (1 - self.error_rate) * 100)
        
        # 根据严重性调整分数
        critical_count = sum(1 for inc in self.inconsistencies if inc.severity == SeverityLevel.CRITICAL)
        if critical_count > 0:
            self.consistency_score *= 0.5  # 严重错误大幅降低分数


class DatabaseConnection:
    """数据库连接管理"""
    
    def __init__(self, db_type: str, config: Dict[str, Any]):
        self.db_type = db_type  # 'mysql' or 'postgresql'
        self.config = config
        self.connection_pool = None
    
    async def initialize(self):
        """初始化连接池"""
        if self.db_type == 'mysql':
            self.connection_pool = await aiomysql.create_pool(**self.config)
        elif self.db_type == 'postgresql':
            self.connection_pool = await asyncpg.create_pool(**self.config)
        else:
            raise ValueError(f"Unsupported database type: {self.db_type}")
        
        logger.info(f"{self.db_type.capitalize()} connection pool initialized")
    
    async def fetch_data(self, sql: str, params: Tuple = None) -> List[Dict[str, Any]]:
        """执行查询并返回数据"""
        if self.db_type == 'mysql':
            async with self.connection_pool.acquire() as conn:
                async with conn.cursor(aiomysql.DictCursor) as cursor:
                    await cursor.execute(sql, params or ())
                    return await cursor.fetchall()
        
        elif self.db_type == 'postgresql':
            async with self.connection_pool.acquire() as conn:
                if params:
                    result = await conn.fetch(sql, *params)
                else:
                    result = await conn.fetch(sql)
                return [dict(row) for row in result]
    
    async def fetch_scalar(self, sql: str, params: Tuple = None) -> Any:
        """执行查询并返回单个值"""
        if self.db_type == 'mysql':
            async with self.connection_pool.acquire() as conn:
                async with conn.cursor() as cursor:
                    await cursor.execute(sql, params or ())
                    result = await cursor.fetchone()
                    return result[0] if result else None
        
        elif self.db_type == 'postgresql':
            async with self.connection_pool.acquire() as conn:
                if params:
                    return await conn.fetchval(sql, *params)
                else:
                    return await conn.fetchval(sql)
    
    async def cleanup(self):
        """清理连接池"""
        if self.connection_pool:
            if self.db_type == 'mysql':
                self.connection_pool.close()
                await self.connection_pool.wait_closed()
            else:
                await self.connection_pool.close()


class DataTypeNormalizer:
    """数据类型标准化器"""
    
    @staticmethod
    def normalize_numeric(value: Any) -> Optional[Decimal]:
        """标准化数值类型"""
        if value is None:
            return None
        
        try:
            if isinstance(value, (int, float)):
                return Decimal(str(value))
            elif isinstance(value, Decimal):
                return value
            elif isinstance(value, str):
                # 处理空字符串
                if value.strip() == '':
                    return Decimal('0')
                return Decimal(value)
            else:
                return Decimal(str(value))
        except Exception:
            return None
    
    @staticmethod
    def normalize_timestamp(value: Any) -> Optional[datetime]:
        """标准化时间戳"""
        if value is None:
            return None
        
        if isinstance(value, datetime):
            return value
        elif isinstance(value, str):
            try:
                # 尝试多种时间格式
                formats = [
                    '%Y-%m-%d %H:%M:%S.%f',
                    '%Y-%m-%d %H:%M:%S',
                    '%Y-%m-%dT%H:%M:%S.%f',
                    '%Y-%m-%dT%H:%M:%S'
                ]
                
                for fmt in formats:
                    try:
                        return datetime.strptime(value, fmt)
                    except ValueError:
                        continue
                
                # ISO格式
                return datetime.fromisoformat(value.replace('Z', '+00:00'))
            except Exception:
                return None
        
        return None
    
    @staticmethod
    def normalize_string(value: Any) -> Optional[str]:
        """标准化字符串"""
        if value is None:
            return None
        
        if isinstance(value, bytes):
            return value.decode('utf-8', errors='ignore')
        
        return str(value).strip()


class DataConsistencyChecker:
    """数据一致性检查器"""
    
    def __init__(self, mysql_config: Dict[str, Any], postgresql_config: Dict[str, Any]):
        self.mysql_conn = DatabaseConnection('mysql', mysql_config)
        self.postgresql_conn = DatabaseConnection('postgresql', postgresql_config)
        self.normalizer = DataTypeNormalizer()
        
        # 表映射配置
        self.table_mappings = {
            'dbtickdata': 'vnpy_tick_history',
            'dbbardata': 'vnpy_bar_history',
            'dborderdata': 'vnpy_order_history',
            'dbtradedata': 'vnpy_trade_history',
            'dbpositiondata': 'vnpy_position_history',
            'dbaccountdata': 'vnpy_account_history'
        }
        
        # 字段映射（MySQL -> PostgreSQL）
        self.field_mappings = {
            'datetime': 'datetime',
            'symbol': 'symbol',
            'exchange': 'exchange',
            'last_price': 'last_price',
            'volume': 'volume',
            'turnover': 'turnover'
            # 可以添加更多字段映射
        }
    
    async def initialize(self):
        """初始化检查器"""
        await self.mysql_conn.initialize()
        await self.postgresql_conn.initialize()
        logger.info("Data consistency checker initialized")
    
    async def check_table_consistency(self, mysql_table: str, 
                                    date_range: Tuple[datetime, datetime] = None,
                                    sample_size: int = None) -> ConsistencyReport:
        """检查表的数据一致性"""
        postgresql_table = self.table_mappings.get(mysql_table, mysql_table)
        
        check_id = f"{mysql_table}_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
        check_start_time = datetime.now()
        
        logger.info(f"Starting consistency check for {mysql_table} -> {postgresql_table}")
        
        try:
            # 1. 检查记录数量一致性
            count_inconsistencies = await self._check_record_counts(
                mysql_table, postgresql_table, date_range
            )
            
            # 2. 检查数据值一致性
            data_inconsistencies = await self._check_data_consistency(
                mysql_table, postgresql_table, date_range, sample_size
            )
            
            # 3. 检查数据类型一致性
            type_inconsistencies = await self._check_data_types(
                mysql_table, postgresql_table
            )
            
            # 4. 检查空值处理一致性
            null_inconsistencies = await self._check_null_handling(
                mysql_table, postgresql_table, date_range
            )
            
            # 合并所有不一致记录
            all_inconsistencies = (
                count_inconsistencies + 
                data_inconsistencies + 
                type_inconsistencies +
                null_inconsistencies
            )
            
            # 获取记录统计
            mysql_count, postgresql_count, total_checked = await self._get_record_statistics(
                mysql_table, postgresql_table, date_range
            )
            
            check_end_time = datetime.now()
            
            report = ConsistencyReport(
                check_id=check_id,
                table_name=mysql_table,
                check_start_time=check_start_time,
                check_end_time=check_end_time,
                total_records_checked=total_checked,
                mysql_record_count=mysql_count,
                postgresql_record_count=postgresql_count,
                inconsistencies=all_inconsistencies
            )
            
            logger.info(f"Consistency check completed: {len(all_inconsistencies)} inconsistencies found")
            return report
            
        except Exception as e:
            logger.error(f"Error in consistency check: {e}")
            logger.error(traceback.format_exc())
            raise
    
    async def _check_record_counts(self, mysql_table: str, postgresql_table: str,
                                 date_range: Tuple[datetime, datetime] = None) -> List[InconsistencyRecord]:
        """检查记录数量一致性"""
        inconsistencies = []
        
        # 构建查询条件
        where_clause = ""
        params = ()
        
        if date_range:
            where_clause = "WHERE datetime >= %s AND datetime < %s"
            params = date_range
        
        # MySQL查询
        mysql_sql = f"SELECT COUNT(*) FROM {mysql_table} {where_clause}"
        mysql_count = await self.mysql_conn.fetch_scalar(mysql_sql, params)
        
        # PostgreSQL查询（调整参数占位符）
        pg_where_clause = where_clause.replace('%s', '$1').replace('%s', '$2') if where_clause else ""
        postgresql_sql = f"SELECT COUNT(*) FROM {postgresql_table} {pg_where_clause}"
        postgresql_count = await self.postgresql_conn.fetch_scalar(postgresql_sql, params)
        
        # 检查数量差异
        if mysql_count != postgresql_count:
            difference = abs(mysql_count - postgresql_count)
            severity = SeverityLevel.CRITICAL if difference > mysql_count * 0.01 else SeverityLevel.HIGH
            
            inconsistencies.append(InconsistencyRecord(
                inconsistency_type=InconsistencyType.MISSING_RECORD,
                severity=severity,
                table_name=mysql_table,
                primary_key="COUNT(*)",
                mysql_value=mysql_count,
                postgresql_value=postgresql_count,
                description=f"Record count mismatch: MySQL({mysql_count}) vs PostgreSQL({postgresql_count})"
            ))
        
        return inconsistencies
    
    async def _check_data_consistency(self, mysql_table: str, postgresql_table: str,
                                    date_range: Tuple[datetime, datetime] = None,
                                    sample_size: int = 1000) -> List[InconsistencyRecord]:
        """检查数据值一致性"""
        inconsistencies = []
        
        # 获取样本数据进行对比
        where_clause = ""
        params = ()
        
        if date_range:
            where_clause = "WHERE datetime >= %s AND datetime < %s"
            params = date_range
        
        # 添加采样
        mysql_sql = f"""
            SELECT * FROM {mysql_table} {where_clause}
            ORDER BY datetime DESC
            LIMIT {sample_size or 1000}
        """
        
        pg_where_clause = where_clause.replace('%s', '$1').replace('%s', '$2') if where_clause else ""
        postgresql_sql = f"""
            SELECT * FROM {postgresql_table} {pg_where_clause}
            ORDER BY datetime DESC
            LIMIT {sample_size or 1000}
        """
        
        # 获取数据
        mysql_data = await self.mysql_conn.fetch_data(mysql_sql, params)
        postgresql_data = await self.postgresql_conn.fetch_data(postgresql_sql, params)
        
        # 创建查找字典（基于主键）
        mysql_dict = {}
        postgresql_dict = {}
        
        for row in mysql_data:
            # 生成主键（symbol + datetime）
            key = f"{row.get('symbol', '')}_{row.get('datetime', '')}"
            mysql_dict[key] = row
        
        for row in postgresql_data:
            key = f"{row.get('symbol', '')}_{row.get('datetime', '')}"
            postgresql_dict[key] = row
        
        # 对比数据
        common_keys = set(mysql_dict.keys()) & set(postgresql_dict.keys())
        
        for key in common_keys:
            mysql_row = mysql_dict[key]
            postgresql_row = postgresql_dict[key]
            
            # 对比每个字段
            field_inconsistencies = await self._compare_record_fields(
                mysql_table, key, mysql_row, postgresql_row
            )
            inconsistencies.extend(field_inconsistencies)
        
        # 检查缺失记录
        mysql_only = set(mysql_dict.keys()) - set(postgresql_dict.keys())
        postgresql_only = set(postgresql_dict.keys()) - set(mysql_dict.keys())
        
        for key in mysql_only:
            inconsistencies.append(InconsistencyRecord(
                inconsistency_type=InconsistencyType.MISSING_RECORD,
                severity=SeverityLevel.HIGH,
                table_name=mysql_table,
                primary_key=key,
                description=f"Record exists in MySQL but not in PostgreSQL"
            ))
        
        for key in postgresql_only:
            inconsistencies.append(InconsistencyRecord(
                inconsistency_type=InconsistencyType.EXTRA_RECORD,
                severity=SeverityLevel.MEDIUM,
                table_name=mysql_table,
                primary_key=key,
                description=f"Record exists in PostgreSQL but not in MySQL"
            ))
        
        return inconsistencies
    
    async def _compare_record_fields(self, table_name: str, primary_key: str,
                                   mysql_row: Dict[str, Any], 
                                   postgresql_row: Dict[str, Any]) -> List[InconsistencyRecord]:
        """对比记录字段"""
        inconsistencies = []
        
        # 重要字段列表
        important_fields = ['last_price', 'open_price', 'high_price', 'low_price', 'close_price', 'volume', 'turnover']
        
        for mysql_field, postgresql_field in self.field_mappings.items():
            if mysql_field not in mysql_row or postgresql_field not in postgresql_row:
                continue
            
            mysql_value = mysql_row[mysql_field]
            postgresql_value = postgresql_row[postgresql_field]
            
            # 数值字段对比
            if mysql_field in important_fields:
                inconsistency = await self._compare_numeric_field(
                    table_name, primary_key, mysql_field, mysql_value, postgresql_value
                )
                if inconsistency:
                    inconsistencies.append(inconsistency)
            
            # 时间字段对比
            elif mysql_field == 'datetime':
                inconsistency = await self._compare_datetime_field(
                    table_name, primary_key, mysql_field, mysql_value, postgresql_value
                )
                if inconsistency:
                    inconsistencies.append(inconsistency)
            
            # 字符串字段对比
            else:
                inconsistency = await self._compare_string_field(
                    table_name, primary_key, mysql_field, mysql_value, postgresql_value
                )
                if inconsistency:
                    inconsistencies.append(inconsistency)
        
        return inconsistencies
    
    async def _compare_numeric_field(self, table_name: str, primary_key: str, field_name: str,
                                   mysql_value: Any, postgresql_value: Any) -> Optional[InconsistencyRecord]:
        """对比数值字段"""
        mysql_decimal = self.normalizer.normalize_numeric(mysql_value)
        postgresql_decimal = self.normalizer.normalize_numeric(postgresql_value)
        
        if mysql_decimal is None and postgresql_decimal is None:
            return None
        
        if mysql_decimal is None or postgresql_decimal is None:
            return InconsistencyRecord(
                inconsistency_type=InconsistencyType.NULL_VALUE_MISMATCH,
                severity=SeverityLevel.HIGH,
                table_name=table_name,
                primary_key=primary_key,
                field_name=field_name,
                mysql_value=mysql_value,
                postgresql_value=postgresql_value,
                description=f"Null value mismatch in {field_name}"
            )
        
        # 计算相对差异
        if mysql_decimal != 0:
            relative_diff = abs(mysql_decimal - postgresql_decimal) / abs(mysql_decimal)
        else:
            relative_diff = abs(postgresql_decimal) if postgresql_decimal != 0 else 0
        
        # 设置容忍度
        tolerance = Decimal('0.000001')  # 0.0001%
        
        if relative_diff > tolerance:
            # 判断严重性
            if relative_diff > Decimal('0.01'):  # 1%
                severity = SeverityLevel.CRITICAL
            elif relative_diff > Decimal('0.001'):  # 0.1%
                severity = SeverityLevel.HIGH
            else:
                severity = SeverityLevel.MEDIUM
            
            return InconsistencyRecord(
                inconsistency_type=InconsistencyType.PRECISION_LOSS,
                severity=severity,
                table_name=table_name,
                primary_key=primary_key,
                field_name=field_name,
                mysql_value=mysql_value,
                postgresql_value=postgresql_value,
                description=f"Numeric precision mismatch in {field_name}: {relative_diff:.6%} difference"
            )
        
        return None
    
    async def _compare_datetime_field(self, table_name: str, primary_key: str, field_name: str,
                                    mysql_value: Any, postgresql_value: Any) -> Optional[InconsistencyRecord]:
        """对比时间字段"""
        mysql_dt = self.normalizer.normalize_timestamp(mysql_value)
        postgresql_dt = self.normalizer.normalize_timestamp(postgresql_value)
        
        if mysql_dt is None and postgresql_dt is None:
            return None
        
        if mysql_dt is None or postgresql_dt is None:
            return InconsistencyRecord(
                inconsistency_type=InconsistencyType.NULL_VALUE_MISMATCH,
                severity=SeverityLevel.HIGH,
                table_name=table_name,
                primary_key=primary_key,
                field_name=field_name,
                mysql_value=mysql_value,
                postgresql_value=postgresql_value,
                description=f"Null datetime mismatch in {field_name}"
            )
        
        # 计算时间差异
        time_diff = abs((mysql_dt - postgresql_dt).total_seconds())
        
        # 容忍1秒的时间差异（考虑精度问题）
        if time_diff > 1:
            severity = SeverityLevel.HIGH if time_diff > 60 else SeverityLevel.MEDIUM
            
            return InconsistencyRecord(
                inconsistency_type=InconsistencyType.TIMESTAMP_DRIFT,
                severity=severity,
                table_name=table_name,
                primary_key=primary_key,
                field_name=field_name,
                mysql_value=mysql_value,
                postgresql_value=postgresql_value,
                description=f"Timestamp drift in {field_name}: {time_diff:.2f}s difference"
            )
        
        return None
    
    async def _compare_string_field(self, table_name: str, primary_key: str, field_name: str,
                                  mysql_value: Any, postgresql_value: Any) -> Optional[InconsistencyRecord]:
        """对比字符串字段"""
        mysql_str = self.normalizer.normalize_string(mysql_value)
        postgresql_str = self.normalizer.normalize_string(postgresql_value)
        
        if mysql_str != postgresql_str:
            return InconsistencyRecord(
                inconsistency_type=InconsistencyType.DATA_MISMATCH,
                severity=SeverityLevel.MEDIUM,
                table_name=table_name,
                primary_key=primary_key,
                field_name=field_name,
                mysql_value=mysql_value,
                postgresql_value=postgresql_value,
                description=f"String mismatch in {field_name}"
            )
        
        return None
    
    async def _check_data_types(self, mysql_table: str, postgresql_table: str) -> List[InconsistencyRecord]:
        """检查数据类型一致性"""
        inconsistencies = []
        
        # 获取MySQL表结构
        mysql_columns = await self.mysql_conn.fetch_data("""
            SELECT COLUMN_NAME, DATA_TYPE, IS_NULLABLE, COLUMN_DEFAULT
            FROM INFORMATION_SCHEMA.COLUMNS
            WHERE TABLE_NAME = %s AND TABLE_SCHEMA = DATABASE()
            ORDER BY ORDINAL_POSITION
        """, (mysql_table,))
        
        # 获取PostgreSQL表结构
        postgresql_columns = await self.postgresql_conn.fetch_data("""
            SELECT column_name, data_type, is_nullable, column_default
            FROM information_schema.columns
            WHERE table_name = $1 AND table_schema = 'public'
            ORDER BY ordinal_position
        """, (postgresql_table,))
        
        # 创建字段映射
        mysql_dict = {col['COLUMN_NAME']: col for col in mysql_columns}
        postgresql_dict = {col['column_name']: col for col in postgresql_columns}
        
        # 检查映射字段的类型一致性
        for mysql_field, postgresql_field in self.field_mappings.items():
            if mysql_field in mysql_dict and postgresql_field in postgresql_dict:
                mysql_col = mysql_dict[mysql_field]
                postgresql_col = postgresql_dict[postgresql_field]
                
                # 检查数据类型兼容性
                if not self._are_types_compatible(mysql_col['DATA_TYPE'], postgresql_col['data_type']):
                    inconsistencies.append(InconsistencyRecord(
                        inconsistency_type=InconsistencyType.TYPE_MISMATCH,
                        severity=SeverityLevel.HIGH,
                        table_name=mysql_table,
                        primary_key=mysql_field,
                        field_name=mysql_field,
                        mysql_value=mysql_col['DATA_TYPE'],
                        postgresql_value=postgresql_col['data_type'],
                        description=f"Data type mismatch: {mysql_col['DATA_TYPE']} vs {postgresql_col['data_type']}"
                    ))
        
        return inconsistencies
    
    def _are_types_compatible(self, mysql_type: str, postgresql_type: str) -> bool:
        """检查数据类型是否兼容"""
        # 类型兼容性映射
        compatible_types = {
            'bigint': ['bigint', 'integer'],
            'int': ['integer', 'bigint'],
            'double': ['numeric', 'double precision', 'real'],
            'varchar': ['character varying', 'varchar', 'text'],
            'text': ['text', 'character varying'],
            'datetime': ['timestamp', 'timestamp without time zone'],
            'timestamp': ['timestamp with time zone', 'timestamp']
        }
        
        mysql_lower = mysql_type.lower()
        postgresql_lower = postgresql_type.lower()
        
        if mysql_lower == postgresql_lower:
            return True
        
        for mysql_base, pg_compatible in compatible_types.items():
            if mysql_lower.startswith(mysql_base):
                return any(postgresql_lower.startswith(pg_type) for pg_type in pg_compatible)
        
        return False
    
    async def _check_null_handling(self, mysql_table: str, postgresql_table: str,
                                 date_range: Tuple[datetime, datetime] = None) -> List[InconsistencyRecord]:
        """检查空值处理一致性"""
        inconsistencies = []
        
        where_clause = ""
        params = ()
        
        if date_range:
            where_clause = "WHERE datetime >= %s AND datetime < %s"
            params = date_range
        
        # 检查重要字段的空值比例
        important_fields = ['last_price', 'volume', 'turnover']
        
        for field in important_fields:
            if field not in self.field_mappings:
                continue
            
            postgresql_field = self.field_mappings[field]
            
            # MySQL空值统计
            mysql_sql = f"""
                SELECT 
                    COUNT(*) as total_count,
                    SUM(CASE WHEN {field} IS NULL OR {field} = 0 THEN 1 ELSE 0 END) as null_count
                FROM {mysql_table} {where_clause}
            """
            mysql_stats = await self.mysql_conn.fetch_data(mysql_sql, params)
            mysql_null_rate = mysql_stats[0]['null_count'] / max(mysql_stats[0]['total_count'], 1)
            
            # PostgreSQL空值统计
            pg_where_clause = where_clause.replace('%s', '$1').replace('%s', '$2') if where_clause else ""
            postgresql_sql = f"""
                SELECT 
                    COUNT(*) as total_count,
                    COUNT(CASE WHEN {postgresql_field} IS NULL OR {postgresql_field} = 0 THEN 1 END) as null_count
                FROM {postgresql_table} {pg_where_clause}
            """
            postgresql_stats = await self.postgresql_conn.fetch_data(postgresql_sql, params)
            postgresql_null_rate = postgresql_stats[0]['null_count'] / max(postgresql_stats[0]['total_count'], 1)
            
            # 检查空值比例差异
            null_rate_diff = abs(mysql_null_rate - postgresql_null_rate)
            
            if null_rate_diff > 0.01:  # 1%的差异阈值
                severity = SeverityLevel.HIGH if null_rate_diff > 0.05 else SeverityLevel.MEDIUM
                
                inconsistencies.append(InconsistencyRecord(
                    inconsistency_type=InconsistencyType.NULL_VALUE_MISMATCH,
                    severity=severity,
                    table_name=mysql_table,
                    primary_key=field,
                    field_name=field,
                    mysql_value=f"{mysql_null_rate:.2%}",
                    postgresql_value=f"{postgresql_null_rate:.2%}",
                    description=f"Null value rate mismatch in {field}: {null_rate_diff:.2%} difference"
                ))
        
        return inconsistencies
    
    async def _get_record_statistics(self, mysql_table: str, postgresql_table: str,
                                   date_range: Tuple[datetime, datetime] = None) -> Tuple[int, int, int]:
        """获取记录统计信息"""
        where_clause = ""
        params = ()
        
        if date_range:
            where_clause = "WHERE datetime >= %s AND datetime < %s"
            params = date_range
        
        # MySQL记录数
        mysql_sql = f"SELECT COUNT(*) FROM {mysql_table} {where_clause}"
        mysql_count = await self.mysql_conn.fetch_scalar(mysql_sql, params)
        
        # PostgreSQL记录数
        pg_where_clause = where_clause.replace('%s', '$1').replace('%s', '$2') if where_clause else ""
        postgresql_sql = f"SELECT COUNT(*) FROM {postgresql_table} {pg_where_clause}"
        postgresql_count = await self.postgresql_conn.fetch_scalar(postgresql_sql, params)
        
        total_checked = max(mysql_count, postgresql_count)
        
        return mysql_count, postgresql_count, total_checked
    
    async def generate_repair_suggestions(self, report: ConsistencyReport) -> List[Dict[str, Any]]:
        """生成修复建议"""
        suggestions = []
        
        # 统计不一致类型
        type_counts = Counter(inc.inconsistency_type for inc in report.inconsistencies)
        
        for inconsistency_type, count in type_counts.items():
            if inconsistency_type == InconsistencyType.MISSING_RECORD:
                suggestions.append({
                    'type': 'data_sync',
                    'priority': 'high',
                    'description': f"Re-sync {count} missing records from MySQL to PostgreSQL",
                    'sql_example': f"""
                        INSERT INTO {self.table_mappings.get(report.table_name, report.table_name)}
                        SELECT * FROM mysql_source WHERE conditions...
                    """
                })
            
            elif inconsistency_type == InconsistencyType.PRECISION_LOSS:
                suggestions.append({
                    'type': 'schema_adjustment',
                    'priority': 'medium',
                    'description': f"Adjust PostgreSQL numeric precision for {count} fields",
                    'sql_example': f"""
                        ALTER TABLE {self.table_mappings.get(report.table_name, report.table_name)}
                        ALTER COLUMN price_field TYPE NUMERIC(28,8);
                    """
                })
            
            elif inconsistency_type == InconsistencyType.TYPE_MISMATCH:
                suggestions.append({
                    'type': 'schema_migration',
                    'priority': 'high',
                    'description': f"Fix {count} data type mismatches",
                    'sql_example': f"""
                        -- Review and update data type mappings
                        -- Consider data migration script
                    """
                })
        
        # 基于一致性分数的建议
        if report.consistency_score < 90:
            suggestions.append({
                'type': 'full_resync',
                'priority': 'critical',
                'description': "Consistency score below 90% - consider full table re-synchronization",
                'action': 'schedule_full_sync'
            })
        
        return suggestions
    
    async def check_all_tables(self, date_range: Tuple[datetime, datetime] = None) -> Dict[str, ConsistencyReport]:
        """检查所有表的一致性"""
        reports = {}
        
        for mysql_table in self.table_mappings.keys():
            try:
                logger.info(f"Checking consistency for table: {mysql_table}")
                report = await self.check_table_consistency(mysql_table, date_range)
                reports[mysql_table] = report
            except Exception as e:
                logger.error(f"Failed to check table {mysql_table}: {e}")
                # 创建错误报告
                reports[mysql_table] = ConsistencyReport(
                    check_id=f"error_{mysql_table}",
                    table_name=mysql_table,
                    check_start_time=datetime.now(),
                    check_end_time=datetime.now(),
                    total_records_checked=0,
                    mysql_record_count=0,
                    postgresql_record_count=0,
                    inconsistencies=[InconsistencyRecord(
                        inconsistency_type=InconsistencyType.DATA_MISMATCH,
                        severity=SeverityLevel.CRITICAL,
                        table_name=mysql_table,
                        primary_key="ERROR",
                        description=f"Check failed: {str(e)}"
                    )]
                )
        
        return reports
    
    async def cleanup(self):
        """清理资源"""
        await self.mysql_conn.cleanup()
        await self.postgresql_conn.cleanup()
        logger.info("Data consistency checker cleanup completed")


# 便捷函数
async def create_consistency_checker(mysql_config: Dict[str, Any], 
                                   postgresql_config: Dict[str, Any]) -> DataConsistencyChecker:
    """创建并初始化一致性检查器"""
    checker = DataConsistencyChecker(mysql_config, postgresql_config)
    await checker.initialize()
    return checker


async def quick_consistency_check(mysql_config: Dict[str, Any], 
                                postgresql_config: Dict[str, Any],
                                table_name: str = None,
                                hours_back: int = 24) -> Dict[str, Any]:
    """快速一致性检查"""
    checker = await create_consistency_checker(mysql_config, postgresql_config)
    
    try:
        # 设置检查时间范围
        end_time = datetime.now()
        start_time = end_time - timedelta(hours=hours_back)
        date_range = (start_time, end_time)
        
        if table_name:
            # 检查单个表
            report = await checker.check_table_consistency(table_name, date_range)
            suggestions = await checker.generate_repair_suggestions(report)
            
            return {
                'report': asdict(report),
                'suggestions': suggestions
            }
        else:
            # 检查所有表
            reports = await checker.check_all_tables(date_range)
            
            # 生成汇总报告
            total_inconsistencies = sum(len(report.inconsistencies) for report in reports.values())
            avg_consistency_score = statistics.mean(report.consistency_score for report in reports.values())
            
            return {
                'summary': {
                    'total_tables_checked': len(reports),
                    'total_inconsistencies': total_inconsistencies,
                    'average_consistency_score': avg_consistency_score,
                    'check_time_range': {
                        'start': start_time.isoformat(),
                        'end': end_time.isoformat()
                    }
                },
                'reports': {table: asdict(report) for table, report in reports.items()}
            }
    
    finally:
        await checker.cleanup()


if __name__ == "__main__":
    async def main():
        """主函数示例"""
        mysql_config = {
            'host': 'localhost',
            'port': 3306,
            'user': 'redfire_user',
            'password': 'your_password',
            'db': 'redfire_business',
            'charset': 'utf8mb4'
        }
        
        postgresql_config = {
            'host': 'localhost',
            'port': 5432,
            'user': 'redfire_analytics',
            'password': 'your_password',
            'database': 'redfire_analytics'
        }
        
        # 快速检查示例
        result = await quick_consistency_check(
            mysql_config, 
            postgresql_config,
            table_name='dbtickdata',
            hours_back=24
        )
        
        print("Consistency Check Results:")
        print(json.dumps(result, indent=2, default=str))
    
    asyncio.run(main())
